use std::{path::{Path, PathBuf}, fs::{OpenOptions, File}, sync::Arc, io::Seek, time::Instant};

use parking_lot::{Mutex, Condvar};

use crate::{constants::{DefaultFingerprintType, XCONTAINER_META_NAME, XCONTAINER_POOL_NAME, XCONTAINER_META_LEN, XCONTAINER_LEN}, chunk::Chunk};

use super::{XcontainerResult, WorkSpace, XcontainerWriter, OffsetMeta, FixedContainer, XcontainerError, XChunkMeta, XcontainerId};

const META_XCONTAINER_SIZE:u64 = (XCONTAINER_LEN * std::mem::size_of::<XChunkMeta>()) as u64;
type XcontainerMetaCollector = FixedContainer::<Arc<FixedContainer<(OffsetMeta,XChunkMeta),XCONTAINER_LEN>>,XCONTAINER_META_LEN>;

pub struct XcontainerWriteManager {
    meta_name:PathBuf,
    pool_name:PathBuf,
    meta_write_position:u64,
    first_id:XcontainerId,
    first_meta:Option<Arc<FixedContainer<(OffsetMeta,XChunkMeta),XCONTAINER_LEN>>>,
    id_allocator:Mutex<XcontainerId>,
    wake_cond:Condvar,
    work_space:Mutex<WorkSpace>,
    write_time:Mutex<u128>,
}

impl XcontainerWriteManager {
    pub fn from_path<P:AsRef<Path>+?Sized>(path:&P) -> XcontainerResult<Self> {
        std::fs::create_dir_all(path).map_err(|e|XcontainerError::XcontainerWriteManager(e))?;
        let meta_name = path.as_ref().join(XCONTAINER_META_NAME);
        let pool_name = path.as_ref().join(XCONTAINER_POOL_NAME);
        let mut meta_file =  OpenOptions::new()
                            .create(true)
                            .read(true)
                            .open(&meta_name)
                            .map_err(|e|XcontainerError::XcontainerMetaFile(e))?;
        let meta_file_size = meta_file
                            .metadata()
                            .map_err(|e|XcontainerError::XcontainerMetaLen(e))?
                            .len();
        let meta_container_num = meta_file_size / META_XCONTAINER_SIZE;
        let meta_file_write_position = meta_container_num * META_XCONTAINER_SIZE;
        let new_meta_container_size = meta_file_size - meta_file_write_position;
        let (first_id,first_meta) = if new_meta_container_size != 0 {
            (meta_container_num,Some(Arc::new(FixedContainer::on_raw_read(&mut meta_file)?)))
        } else {
            (meta_container_num,None)
        };
        Ok(Self {
            meta_name:meta_name,
            pool_name:pool_name,
            meta_write_position:meta_file_write_position,
            first_id:first_id,
            id_allocator:Mutex::new(first_id),
            first_meta:first_meta,
            wake_cond:Condvar::new(),
            work_space:Mutex::new(WorkSpace::new()),
            write_time:Mutex::new(0),
        })
    }

    pub fn find_in_work_space(&self,fp:&DefaultFingerprintType) -> Option<Arc<Chunk>> {
        self.work_space.lock().find_in_work_space(fp)
    }

    pub fn insert_to_recent_buffer(&self,chunk:Arc<Chunk>) {
        self.work_space.lock().insert_to_recent_buffer(chunk);
    }

    pub fn insert_to_write_buffer(&self,xcon:Arc<XcontainerWriter>) {
        self.work_space.lock().insert_to_write_buffer(xcon);
    }

    pub fn next_write_xcon(&self) -> Option<Arc<XcontainerWriter>> {
        let mut work_space = self.work_space.lock();
        let result = work_space.next_write_xcon();
        if !work_space.enable() && result.is_none() {
            self.wake_cond.wait(&mut work_space);
            work_space.next_write_xcon()
        } else {
            result
        }
    }

    pub fn current_xcon(&self) -> Option<Arc<XcontainerWriter>> {
        self.work_space.lock().current_xcon()
    }

    pub fn disable(&self) {
        self.work_space.lock().disable();
        self.wake_cond.notify_all();
    }

    pub fn alloc_xcontainer(&self) -> Arc<XcontainerWriter> {
        let mut cur_id = self.id_allocator.lock();
        if *cur_id == self.first_id {
            if let Some(first_meta) = &self.first_meta {
                let result = Arc::new(XcontainerWriter::half(*cur_id, first_meta.clone()));
                *cur_id += 1;
                return result;
            }
        }
        let result = Arc::new(XcontainerWriter::new(*cur_id));
        *cur_id += 1;
        result
    }

    pub fn write_process(&self) -> XcontainerResult<()> {
        let mut xcon_meta_container = XcontainerMetaCollector::new();
        let mut offset_vec = Vec::new();
        let mut meta_file = OpenOptions::new()
                            .create(true)
                            .write(true)
                            .open(&self.meta_name)
                            .map_err(|e|XcontainerError::XcontainerMetaFile(e))?;
        let mut pool_file = OpenOptions::new()
                            .create(true)
                            .write(true)
                            .open(&self.pool_name)
                            .map_err(|e|XcontainerError::XcontainerPoolFile(e))?;
        meta_file
            .seek(std::io::SeekFrom::Start(self.meta_write_position))
            .map_err(|e|XcontainerError::XcontainerMetaFile(e))?;
        pool_file
            .seek(std::io::SeekFrom::End(0))
            .map_err(|e|XcontainerError::XcontainerPoolFile(e))?;
        let mut cost_time = 0;
        loop {
            let xcon = self.next_write_xcon();
            if let Some(xcon) = xcon {
                let now = Instant::now();
                let (offset,container_meta) = xcon.on_write(&mut pool_file)?;
                if let Err(e) = xcon_meta_container.append_arc_item(&container_meta) {
                    if let XcontainerError::FullContainer = e {
                        xcon_meta_container.on_vec_write(&mut meta_file)?;
                        xcon_meta_container = XcontainerMetaCollector::new();
                        xcon_meta_container.append_arc_item(&container_meta)?;
                    } else {
                        return Err(e);
                    }
                }
                offset_vec.push(offset);
                cost_time += now.elapsed().as_micros();
            } else {
                break;
            }
        }
        if xcon_meta_container.len() != 0 {
            xcon_meta_container.on_vec_write(&mut meta_file)?;
        }
        *self.write_time.lock() = cost_time;
        Ok(())
    }
}

pub struct XcontainerReadManager {
    _meta:File,
    _pool:File,
}

impl XcontainerReadManager {
    pub fn from_path<P:AsRef<Path>+?Sized>(path:&P) -> XcontainerResult<Self> {
        let meta_name = path.as_ref().join(XCONTAINER_META_NAME);
        let pool_name = path.as_ref().join(XCONTAINER_POOL_NAME);
        let meta = OpenOptions::new()
                            .read(true)
                            .open(meta_name)
                            .map_err(|e|XcontainerError::XcontainerReadManager(e))?;
        let pool = OpenOptions::new()
                            .read(true)
                            .open(pool_name)
                            .map_err(|e|XcontainerError::XcontainerReadManager(e))?;
        Ok(Self {
            _meta:meta,
            _pool:pool,
        })
    }
}